/*
 * ManagerImpl.h
 *
 *  Created on: 2011/06/04
 *      Author: naoki
 */

#ifndef INCLUDED_NTENGINE_THREADING_MANAGERIMPL_H
#define INCLUDED_NTENGINE_THREADING_MANAGERIMPL_H

#include "NTEngine/Threading/Manager.h"
#include "NTEngine/Threading/Semaphore.h"
#include "NTEngine/Threading/Mutex.h"
#include "NTEngine/Threading/Thread.h"
#include "NTEngine/Threading/Event.h"
#include <unistd.h>
#include <pthread.h>
#include <list>

namespace NTEngine {
namespace Threading {

class ManagerImpl {
public:
	ManagerImpl( int addThreadNumber )
	: mThreads( NULL )
	, mThreadNumber( 0 )
	, mCoreNumber( 1 )
	, mEndRequest( false ) {
		mQueueSemaphore = Semaphore::create( 0, 0x7fffffff );
		mLock = Mutex::create();

		// CPUコア数取得
		mCoreNumber = sysconf( _SC_NPROCESSORS_CONF );

		mThreadNumber = addThreadNumber + mCoreNumber + 1;
		mThreads = NEW pthread_t[ mThreadNumber ];

		for ( int i = 0; i < mThreadNumber; ++i ) {
			pthread_create( &mThreads[ i ], NULL, entry, this );
		}
	}

	~ManagerImpl() {
		mLock.lock();
		mEndRequest = true;
		mLock.unlock();

		for ( int i = 0; i < mThreadNumber; ++i ) {
			while ( 1 ) {
				mQueueSemaphore.increase();
				// デタッチスレッドにして、無理やり閉じる。joinはたまに動作しない・・・
				int r = pthread_detach( mThreads[ i ] );
				//int r = pthread_join( mThreads[ i ], 0 );
				if ( r == 0 ) {
					mThreads[ i ] = 0;
					break;
				}
			}
		}
		SAFE_DELETE_ARRAY( mThreads );
	}

	static void* entry( void* self ) {
		static_cast< ManagerImpl* >( self )->execute();
		return NULL;
	}

	void execute() {
		while ( 1 ) {
			mQueueSemaphore.decrease();

			mLock.lock();
			if ( mEndRequest ) {
				mLock.unlock();
				break;
			}
			Entry e = mRequestQ.front();
			mRequestQ.pop_front();
			mLock.unlock();
			e.mThread->operator()();
			e.mFinished->set();
		}
	}

	void add( Thread* t, Event* e ) {
		mLock.lock();
		mRequestQ.push_back( Entry( t, e ) );
		mLock.unlock();
		mQueueSemaphore.increase();
	}

	int getCoreNumber() const {
		return mCoreNumber;
	}

	class Entry {
	public:
		Entry()
		: mThread( NULL )
		, mFinished( NULL ) {
		}

		Entry( Thread* t, Event* e )
		: mThread( t )
		, mFinished( e ) {
		}

		~Entry() {
			mThread = NULL;
			mFinished = NULL;
		}

		Thread* mThread;
		Event* mFinished;
	};
	pthread_t* mThreads;
	int mThreadNumber;
	int mCoreNumber;
	bool mEndRequest;
	Semaphore mQueueSemaphore;
	Mutex mLock;
	std::list< Entry > mRequestQ;
};

extern ManagerImpl* gManagerImpl;

} // namespace Threading
} // namespace NTEngine

#endif // INCLUDED_NTENGINE_THREADING_MANAGERIMPL_H
